package com.atguigu.edu.util;

import com.alibaba.fastjson.JSONObject;

import com.atguigu.edu.bean.TableProcess;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;


public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    
    private Connection conn;
//    private Jedis redisClient;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        conn = JdbcUtil.getPhoenixConnection();

//        redisClient = RedisUtil.getRedisClient();
    }
    
    @Override
    public void close() throws Exception {
        JdbcUtil.closeConnection(conn);

//        redisClient.close();
        
    }
    
    //一条数据执行一次
    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> t,
                       Context ctx) throws Exception {
        JSONObject data = t.f0;
        TableProcess tp = t.f1;
        // 1. 把维度写入到 phoenix 中
        writeDimToPhoenix(data, tp);
        // 2. 删除缓存(相比更新, 更加高效: 只需要链接 redis 一次)
//        delCache(data, tp);
    }
    
    private void delCache(JSONObject data, TableProcess tp) {
        String key = DimUtil.getKey(tp.getSinkTable(), data.getString("id"));
//        redisClient.del(key);
    }
    
    private void writeDimToPhoenix(JSONObject data, TableProcess tp) throws SQLException {
        // 使用 jdbc 把数据写入 phoenix 中
        // 1. 拼接一个 sql 插入语句
        // upsert into user(a,b,c)values(?,?,?)
        StringBuilder sql = new StringBuilder();
        sql
            .append("upsert into ")
            .append(tp.getSinkTable())
            .append("(")
            // 拼接字段
            .append(tp.getSinkColumns())
            .append(")values(")
            // 拼接占位符
            .append(tp.getSinkColumns().replaceAll("[^,]+", "?"))
            .append(")");
        
        System.out.println("phoenix 插入语句: " + sql);
        // 2. 获取预处理
        PreparedStatement ps = conn.prepareStatement(sql.toString());
        // 3. 占位符赋值
        String[] columns = tp.getSinkColumns().split(",");
        for (int i = 0; i < columns.length; i++) {
            String columnName = columns[i];
            String v = data.getString(columnName);
            ps.setString(i + 1, v);
        }
        
        // 4. 执行预处理
        ps.execute();
        // 5. 提交
        conn.commit();
        // 6. 关闭预处理
        ps.close();
    }
}
